package com.elvin.platform.rabitMQ.consumeService;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.elvin.platform.beans.Device;
import com.elvin.platform.beans.HistoryData;
import com.elvin.platform.service.DeviceService;
import com.elvin.platform.service.HistoryDataService;
import com.elvin.platform.webSocket.BaseWebsocketClient;
import io.socket.client.IO;
import io.socket.client.Socket;
import org.java_websocket.enums.ReadyState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

@Component
public class RabbitmqConsumer {

  private static Logger logger = LoggerFactory.getLogger(RabbitmqConsumer.class);
  @Autowired
  private RabbitTemplate rabbitTemplate;
  @Autowired
  private DeviceService deviceService;
  @Autowired
  private HistoryDataService historyDataService;
  @RabbitListener(queues = "recived-queue")
  @RabbitHandler
  public void getRabbitmqMessage(Message message){
    //字节码转化为对象
    //转换成为JSONObject对象
    SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    byte[] bytes=message.getBody();
    String msg = new String(bytes);
    try {
      ReceiveVo receiveVo = JSONObject.parseObject(msg,ReceiveVo.class);
      logger.info("RabbitMQ收到的消息为："+receiveVo.toString());
      //收到消息之后进行其他数据处理
      //1.根据 dev_plateform=='gateway' 和  dev_code=='devieceCode'查找设备
      // 修改设备表中的设备，将reportTime和reportdata值放到report1中
      EntityWrapper<Device> wrapper = new EntityWrapper<>();
      wrapper.eq("dev_platform", "gateway");
      wrapper.eq("dev_code", receiveVo.getDevieceCode());
      List<Device> deviceList = deviceService.selectList(wrapper);
      JSONObject jsonObject = new JSONObject();
      jsonObject.put("report_time",receiveVo.getReportTime());
      jsonObject.put("data",receiveVo.getReportData());
      if (deviceList.size()>0) {
        Device device1 = new Device();
        device1.setReport1(jsonObject.toString());
        for (Device device : deviceList) {
          device1.setId(device.getId());
          device1.setUpdateTime(new Date());
          deviceService.updateById(device1);
        }
      }
      //2.新增到 report_data表，新dev_code,report,repoet_date,type
      HistoryData historyData = new HistoryData();
      historyData.setDevCode(receiveVo.getDevieceCode());
      historyData.setReport(jsonObject.toString());
      historyData.setReportDate(formatter.parse(receiveVo.getReportTime()));
      historyData.setDataSources("gateway_pressure");
      historyData.setType(receiveVo.getType());
      historyData.setCreateTime(new Date());
      historyDataService.insert(historyData);

      //3、发送消息到webSocket服务器
      logger.info("rabbitMQ发送消息到webSocket开始：");
      BaseWebsocketClient myClient = new BaseWebsocketClient(new URI("ws://81.68.121.222:8006/platform/api/pushMessage"));
      //BaseWebsocketClient myClient = new BaseWebsocketClient(new URI("ws://localhost:8006/platform/api/pushMessage"));
      myClient.connect();
      while (!myClient.getReadyState().equals(ReadyState.OPEN)) {
        System.out.println("连接中。。。");
        Thread.sleep(1000);
      }
      JSONObject object = new JSONObject();
      object.put("message", "success连接");
      myClient.send(msg);
      logger.info("rabbitMQ发送消息到webSocket结束：");
      //发送数据到socket.io-client
     /* System.out.println("发送数据到socket.io-client开始");
      sendToSocketIOService(msg);
      System.out.println("发送数据到socket.io-client结束");*/
    } catch (Exception e) {
      logger.info(e.getMessage());
    }
  }


  public void sendMessage(String message) {
    System.out.println("发送的消息是：" + message);
    rabbitTemplate.convertAndSend("test_exchange", "D#.T#.P#", message);
  }


  public void sendToSocketIOService(String msg){
    String url = "http://81.68.121.222:7001";
    try {
      IO.Options options = new IO.Options();
      options.transports = new String[]{"websocket"};
      //失败重试次数
      options.reconnectionAttempts = 10;
      //失败重连的时间间隔
      options.reconnectionDelay = 1000;
      //连接超时时间(ms)
      options.timeout = 500;
      final Socket socket = IO.socket(url, options);
      //监听自定义msg事件
      socket.on("res", objects -> System.out.println("client: 服务端发送的消息->" + Arrays.asList(objects)));
      //监听自定义订阅事件
      socket.on("msg", objects -> System.out.println("client: " + "订阅成功，收到反馈->" + Arrays.asList(objects)));

      socket.on(Socket.EVENT_CONNECT, objects -> {
        socket.emit("msg","你好，socket！");
        System.out.println("client: " + "连接成功");
      });
      socket.on(Socket.EVENT_CONNECTING, objects -> System.out.println("client: " + "连接中"));
      socket.on(Socket.EVENT_CONNECT_TIMEOUT, objects -> System.out.println("client: " + "连接超时"));
      socket.on(Socket.EVENT_CONNECT_ERROR, objects -> System.out.println("client: " + "连接失败"));
      socket.connect();
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }


}
